package com.txw.javaCore.server.config;

import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.util.HashMap;
import java.util.Map;

/**
 * @PROJECT_NAME: JavaCoreB2
 * @DESCRIPTION:
 * @Author: 涂玄武
 * @DATE: 2021/1/5 14:00
 */
@Configuration
public class RabbitConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitConfig.class);

    @Autowired
    private Environment env;

    //自动设置RabbitMQ的链接工厂实例
    @Autowired
    private CachingConnectionFactory connectionFactory;

    //自动设置消息监听器所在的容器工厂配置类实例
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * Rabbitmq自定义注入模板操作组件
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String s) {
                log.info("消息发送成功：correlationData={},ack={},s={}", correlationData, ack, s);
            }
        });

        //当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法
        //将消息返还给生产者;
        //当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;
        //通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                log.info("消息没有成功路由到绑定的队列：message={},i={},s={},s1={},s2={}", message, i, s, s1, s2);
            }
        });

        return rabbitTemplate;
    }

    //单实例消费者
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        return factory;
    }

    //多实例消费者
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //确认消费模式-NONE
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.concurrency", int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.max-concurrency", int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.simple.prefetch", int.class));
        return factory;
    }

    /**
     * 创建死信队列的消息模型（邮件定时发送）
     * 死信队列：死信交换机+死信路由（TTL存活时间）
     *
     * @return
     */
    @Bean
    public Queue orderDeadQueue() {
        //创建死信队列的组成成分Map,用于存放组成成分的相关成员
        HashMap<String, Object> argMap = Maps.newHashMap();
        //创建死信交换机
        argMap.put("x-dead-letter-exchange", env.getProperty("mq.order.dead.exchange"));
        //创建死信路由
        argMap.put("x-dead-letter-routing-key", env.getProperty("mq.order.dead.routing.key"));
        //设定TTL,单位为ms，在这里指的是10s（当队列和消息同时设置了TTL，则消息的最大生存时间或者存活时间将取两者中较短的时间）
        //argMap.put("x-message-ttl",10000);
        return new Queue(env.getProperty("mq.order.dead.queue"), true, false, false, argMap);
    }

    /**
     * 创建真正监听队列，面向消费者
     * @return
     */
    @Bean
    public Queue orderBasicQueue() {
        //创建并返回面向消费者的真正队列实例
        return new Queue(env.getProperty("mq.order.basic.queue"), true);
    }

    /**
     * 创建死信交换机
     * @return
     */
    @Bean
    public TopicExchange orderDeadExchange() {
        //创建并返回死信交换机实例
        return new TopicExchange(env.getProperty("mq.order.dead.exchange"), true, false);
    }

    /**
     * 将死信交换机+死信路由绑定到真正的监听队列
     * @return
     */
    @Bean
    public Binding orderDeadBingding(){
        //创建死信路由及其绑定实例
        return BindingBuilder.bind(orderBasicQueue()).to(orderDeadExchange()).with(env.getProperty("mq.order.dead.routing.key"));
    }

    /**
     * 创建”基本消息模型“的基本交换机，面向生产者
     * @return
     */
    @Bean
    public TopicExchange orderBasicExchange(){
        return new TopicExchange(env.getProperty("mq.order.basic.exchange"),true,false);
    }

    /**
     * 创建”基本消息模型“的基本绑定（基本交换机+基本路由）绑定到死信队列，面向生产者
     * @return
     */
    @Bean
    public Binding orderBasicBinding(){
        return BindingBuilder.bind(orderDeadQueue()).to(orderBasicExchange()).with(env.getProperty("mq.order.basic.routing.key"));
    }

    /**
     * 创建延时队列的消息模型（邮件定时发送）
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return QueueBuilder.durable(env.getProperty("mq.delay.queue")).build();
    }

    @Bean
    public CustomExchange delayExchange(){
        Map<String,Object> map=Maps.newHashMap();
        map.put("x-delayed-type","direct");
        return new CustomExchange(env.getProperty("mq.delay.exchange"),"x-delayed-message",true,false,map);
    }

    @Bean
    public Binding delayBinding(){
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(env.getProperty("mq.delay.routing.key")).noargs();
    }

    //创建用户下单死信队列的消息模型
    //死信队列
    @Bean
    public Queue orderDeadQueueV2(){
        Map<String, Object> argMap = Maps.newHashMap();
        argMap.put("x-dead-letter-exchange",env.getProperty("mq.order.dead.exchange.v2"));
        argMap.put("x-dead-letter-routing-key",env.getProperty("mq.order.dead.routing.key.v2"));
        return new Queue(env.getProperty("mq.order.dead.queue.v2"),true,false,false,argMap);
    }

    //死信交换机
    @Bean
    public TopicExchange orderDeadExchangeV2(){
        return new TopicExchange(env.getProperty("mq.order.dead.exchange.v2"),true,false);
    }

    //真正的队列
    @Bean
    public Queue orderBasicQueueV2(){
        return new Queue(env.getProperty("mq.order.basic.queue.v2"),true);
    }

    //死信交换机+死信路由绑定到->真正的队列
    @Bean
    public Binding orderDeadBindingV2(){
        return BindingBuilder.bind(orderBasicQueueV2()).to(orderDeadExchangeV2()).with(env.getProperty("mq.order.dead.routing.key.v2"));
    }

    //基本交换机
    @Bean
    public TopicExchange orderBasicExchangeV2(){
        return new TopicExchange(env.getProperty("mq.order.basic.exchange.v2"),true,false);
    }

    //基本交换机+基本路由绑定到->死信队列
    @Bean
    public Binding orderBasicBindingV2(){
        return BindingBuilder.bind(orderDeadQueueV2()).to(orderBasicExchangeV2()).with(env.getProperty("mq.order.basic.routing.key.v2"));
    }

    //用户登录成功写日志消息模型创建
    /**
     * 创建队列
     * @return
     */
    @Bean(name = "loginQueue")
    public Queue loginQueue(){
        return new Queue(env.getProperty("mq.login.queue.name"),true);
    }

    /**
     * 创建交换机
     * @return
     */
    @Bean
    public TopicExchange loginExchange(){
        return new TopicExchange(env.getProperty("mq.login.exchange.name"),true,false);
    }

    /**
     * 创建绑定
     * @return
     */
    @Bean
    public Binding loginBinding(){
        return BindingBuilder.bind(loginQueue()).to(loginExchange()).with(env.getProperty("mq.login.routing.key.name"));
    }
}
